package com.xdcplus.web.service.impl;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.xdcplus.web.common.pojo.entity.DwWorkConf;
import com.xdcplus.web.common.pojo.entity.crack.TpCrackBsgTest;
import com.xdcplus.web.common.pojo.vo.crack.TpCrackTestVO;
import com.xdcplus.web.mapper.TpCrackMapper;
import com.xdcplus.web.service.TpCrackService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import java.sql.Timestamp;
import java.time.format.DateTimeFormatter;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;


@Slf4j
@Service
public class TpCrackServiceImpl
        extends ServiceImpl<TpCrackMapper, TpCrackBsgTest>
        implements TpCrackService {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    private String MES_TABLE_DML_HIS_TOPIC = "tp_crack_test_to_doris";

    public  static final int KafkaSendMessageMaxSize = 10;

    @Override
    public void syncData(DwWorkConf dwWorkConf) {

        List<TpCrackTestVO> tpCrackData = queryBsgAndPsgData(
                dwWorkConf.getEtlStartTimestamp(),
                dwWorkConf.getEtlEndTimestamp()
        );

        if (ObjectUtil.isNotEmpty(tpCrackData)) {
            List<Map<String, Object>> transformedList = tpCrackData.stream()
                    .filter(Objects::nonNull)
                    .map(tracePrdInfo -> {
                        Map<String, Object> objectMap = JSON.parseObject(JSON.toJSONString(tracePrdInfo)).getInnerMap()
                                .entrySet()
                                .stream()
                                .collect(
                                        HashMap::new,
                                        (map, item) -> map.put(
                                                StrUtil.toUnderlineCase(item.getKey()).toUpperCase(), item.getValue()),
                                        HashMap::putAll
                                );
                        return objectMap;
                    })
                    .collect(Collectors.toList());

            /// 发送消息大小限制
            if (CollectionUtil.isNotEmpty(transformedList)) {
                log.info("清洗数量: {}",transformedList.size());
                AtomicInteger atomicInteger = new AtomicInteger(0);
                while (atomicInteger.get() < transformedList.size()) {
                    int toIndex = atomicInteger.get() + KafkaSendMessageMaxSize;
                    List<Map<String, Object>> subList = transformedList.subList(atomicInteger.get(), toIndex > transformedList.size() ? transformedList.size() : toIndex);
                    atomicInteger.set(toIndex);
                    kafkaTemplate.send(MES_TABLE_DML_HIS_TOPIC, JSON.toJSONString(subList));
                }
            }
        }else {
            log.info("bsg/psg同步数据任务， {} 查询数据为空！",dwWorkConf);
        }

    }

    @Override
    public List<TpCrackTestVO> queryBsgAndPsgData(Timestamp startTime, Timestamp endTime) {
        List<TpCrackBsgTest> tpCrackBsgTests = baseMapper.queryBsgData(startTime, endTime);
        tpCrackBsgTests.addAll(baseMapper.queryPsgData(startTime, endTime)) ;
        return tpCrackBsgTests.stream().map(
                v->{
                    TpCrackTestVO tpCrackTestVO = BeanUtil.copyProperties(v, TpCrackTestVO.class);
                    tpCrackTestVO.setCreatedTime(v.getCreatedTime().toLocalDateTime().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
                    tpCrackTestVO.setDetectTime(v.getDetectTime().toLocalDateTime().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
                    return tpCrackTestVO;
                }
        ).collect(Collectors.toList());
    }


}
